package com.example.processfunction;

import com.example.beans.Event;
import com.example.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;


/**
 *  
 *  * @projectName flink-1.13.2-study
 *  * @title     ProcessFunctionExample   
 *  * @package    com.example.processfunction  
 *  * @description    ProcessFunction 处理函数示例  
 *  * @author hjx     
 *  * @date   2022-3-29 15:31  
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *  
 */
public class ProcessFunctionExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event event, long l) {
                                        return event.timestamp;
                                    }
                                })
                )
                .process(new MyProcessFuntion())
                .print();

        env.execute();
    }


    private static class MyProcessFuntion extends ProcessFunction<Event, String> {

        // 这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务”TimerService 来注册的。
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
//            super.onTimer(timestamp, ctx, out);
        }

        /*
            value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类型一致。
            ctx：类型是 ProcessFunction 中定义的内部抽象类 Context，表示当前运行的
                上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服
                务”(TimerService)，以及可以将数据发送到“侧输出流”（side output）的方法.output()。
            out：“收集器”（类型为 Collector），用于返回输出数据。使
         */
        @Override
        public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
            if (value.user.equals("Mary")) {
                out.collect(value.user);
            } else if (value.user.equals("Bob")) {
                out.collect(value.user);
                out.collect(value.user);
            }
            System.out.println(ctx.timerService().currentWatermark());
        }
    }
}
